/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.nifi.processors.huawei.obs;

import com.obs.services.ObsClient;
import com.obs.services.ObsConfiguration;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processors.huawei.credentials.provider.service.HuaweiCredentialsProviderService;
import org.apache.nifi.proxy.ProxyConfiguration;
import org.apache.nifi.proxy.ProxyConfigurationService;
import org.apache.nifi.ssl.SSLContextService;

import javax.net.ssl.SSLContext;
import java.net.Proxy;
import java.util.concurrent.TimeUnit;

import static org.apache.nifi.processors.huawei.common.PropertyDescriptors.*;

public class OBSUtils {
    public static String getUrl(String region,String endpoint, String bucket, String objectKey) {
        if (StringUtils.isBlank(region) || StringUtils.isBlank(bucket) || StringUtils.isBlank(objectKey)) {
            return null;
        }

        if (objectKey.startsWith("/")) {
            objectKey = objectKey.substring(1);
        }

        if (!OBSRegions.NOT_FOUND.equals(OBSRegions.fromName(region))) {
            endpoint = OBSRegions.fromName(region).getEndpoint();
        }
        return String.format("http://%s.%s/%s", bucket, endpoint, objectKey);
    }

    public static ObsClient createObsClientWithAkSk(ProcessContext context, ComponentLog logger) {
        final String accessKey = context.getProperty(ACCESS_KEY).evaluateAttributeExpressions().getValue();
        final String secretKey = context.getProperty(SECRET_KEY).evaluateAttributeExpressions().getValue();
        ObsClient obsClient = new ObsClient(accessKey, secretKey, createConfiguration(context));
        if (logger != null) {
            logger.warn("create ObsClient success");
        }
        return obsClient;
    }

    /**
     * Attempts to create the client using the controller service first before falling back to the standard configuration.
     *
     * @param context The process context
     * @return The created client
     */
    public static ObsClient createClient(final ProcessContext context, ComponentLog logger) {
        final ControllerService service = context.getProperty(HUAWEI_CREDENTIALS_PROVIDER_SERVICE).asControllerService();
        if (service != null) {
            logger.debug("Using Huawei credentials provider service for creating client");
            final HuaweiCredentialsProviderService huaweiCredentialsProviderService =
                    context.getProperty(HUAWEI_CREDENTIALS_PROVIDER_SERVICE).asControllerService(HuaweiCredentialsProviderService.class);
            return new ObsClient(huaweiCredentialsProviderService.getCredentialsProvider(), createConfiguration(context));
        } else {
            logger.debug("Using Huawei credentials for creating client");
            return createObsClientWithAkSk(context, logger);
        }
    }

    private static ObsConfiguration createConfiguration(final ProcessContext context) {
        final ObsConfiguration config = new ObsConfiguration();

        config.setMaxConnections(context.getMaxConcurrentTasks());
        config.setMaxErrorRetry(0);
        // If this is changed to be a property, ensure other uses are also changed
        final int commsTimeout = context.getProperty(TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue();
        config.setConnectionTimeout(commsTimeout);
        config.setSocketTimeout(commsTimeout);
        // not set
        if(context.getProperty(SSL_CONTEXT_SERVICE).isSet()) {
            final SSLContextService sslContextService = context.getProperty(SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class);
            if (sslContextService != null) {
                final SSLContext sslContext = sslContextService.createContext();
                config.setSslProvider(sslContext.getProvider().getName());
            }
        }

        final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(context, () -> {
            if (context.getProperty(ProxyConfigurationService.PROXY_CONFIGURATION_SERVICE).isSet()) {
                final ProxyConfigurationService configurationService = context.getProperty(ProxyConfigurationService.PROXY_CONFIGURATION_SERVICE).asControllerService(ProxyConfigurationService.class);
                return configurationService.getConfiguration();
            }
            return ProxyConfiguration.DIRECT_CONFIGURATION;
        });

        if (Proxy.Type.HTTP.equals(proxyConfig.getProxyType())) {
            config.setHttpProxy(proxyConfig.getProxyServerHost(), proxyConfig.getProxyServerPort(), proxyConfig.getProxyUserName(), proxyConfig.getProxyUserPassword());
        }

        String name = context.getProperty(OBS_REGION).getValue();
        OBSRegions region = OBSRegions.fromName(name);
        String endpoint = context.getProperty(ENDPOINT_OVERRIDE_URL).getValue();
        if (!OBSRegions.NOT_FOUND.equals(region)) {
            endpoint = region.getEndpoint();
        }
        config.setEndPoint(endpoint);
        return config;
    }
}
